package data.platform.cassandra.infra.persistence.repository;

import com.datastax.oss.driver.api.core.CqlSession;
import data.platform.cassandra.infra.persistence.mapping.DataPointEO;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Objects;

@ConditionalOnBean(name = "cassandraConfig")
@Repository
@AllArgsConstructor
@Slf4j
public class CassandraDataPointRepository {

    final ReactiveCassandraTemplate reactiveCassandraTemplate;

    final CqlSession cqlSession;

    private static String CREATE_SQL = "INSERT INTO data_point (metric, tag_json, day, offset, value,event_value) VALUES (?, ?, ?, ?, ?,?)";

    @PostConstruct
    public void init() {
        reactiveCassandraTemplate.setApplicationEventPublisher(null);
    }

    public Mono<DataPointEO> insertRaw(DataPointEO entity) {
        return reactiveCassandraTemplate.getReactiveCqlOperations().execute(CREATE_SQL, entity.getDataPointKey().getMetric()
                , entity.getDataPointKey().getTagJson()
                , entity.getDataPointKey().getPartition()
                , entity.getDataPointKey().getOffset()
                , entity.getValue()
                , entity.getEventValue())
                .map(result -> entity);
    }

    public Flux<DataPointEO> insertAll(List<DataPointEO> entities) {
        return Flux.fromIterable(entities).flatMap(entity -> insertRaw(entity));
    }

    public Flux<DataPointEO> findBySql(Flux<String> sqlFlux) {
        return sqlFlux.flatMap(sql -> reactiveCassandraTemplate.select(sql, DataPointEO.class));
    }

    public Flux<DataPointEO> find(Flux<GroupedFlux<Integer, String>> groups) {
        return groups.flatMap(groupFlux -> findBySql(groupFlux))
                .filter(eo -> Objects.nonNull(eo.getDataPointKey().getMetric()));
    }

}
